package com.chenzhiling.fs.objectStorage;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;
import org.javaswift.joss.client.factory.AccountFactory;
import org.javaswift.joss.client.factory.AuthenticationMethod;
import org.javaswift.joss.model.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * @Author: CHEN ZHI LING
 * @Date: 2021/9/27
 * @Description: ceph对象存储文件系统
 * 继承自hadoop fileSystem类使用joss包实现
 */
public class CephObjectSystem extends FileSystem {

    public static final Logger LOG = LoggerFactory.getLogger(CephObjectSystem.class);
    //ceph平台用户名
    public static final String FS_CEPH_USERNAME = "fs.ceph.username";
    //ceph平台密码
    public static final String FS_CEPH_PASSWORD = "fs.ceph.password";
    //ceph平台网址
    public static final String FS_CEPH_URI = "fs.ceph.uri";
    //登录账号
    public Account account = null;
    //入参uri
    private URI uri;


    @Override
    public String getScheme() {
        return "ceph";
    }


    /**
     * 初始化连接
     * @param cephUri uri
     * @param conf 参数配置
     * @throws IOException 连接错误
     */
    @Override
    public void initialize(URI cephUri, Configuration conf) throws IOException{
        this.setConf(conf);
        this.uri = cephUri;
        this.account = getAccount();
        super.initialize(cephUri,conf);
    }


    //创建链接
    private   Account getAccount(){
        Configuration conf = getConf();
        return new AccountFactory().setUsername(conf.get(FS_CEPH_USERNAME))
                .setPassword(conf.get(FS_CEPH_PASSWORD))
                .setAuthenticationMethod(AuthenticationMethod.BASIC)
                .setAuthUrl(conf.get(FS_CEPH_URI)).createAccount();
    }

    /**
     * 返回传入进来的uri
     * ceph:///桶名/对象名/
     * @return uri
     */
    @Override
    public URI getUri() {
        return uri;
    }

    @Override
    public FSDataInputStream open(Path path, int i) throws IOException {
        CephObjectAttributes cephFiLe = CephObjectUtil.getCephFiLeByPath(path);
        //获得桶
        Container container = account.getContainer(cephFiLe.getBucketName());
        if (!container.exists()){
            throw new FileNotFoundException("container is not existed");
        }
        //获得对象
        StoredObject object = container.getObject(cephFiLe.getObjectName());
        if (!object.exists()){
            throw new FileNotFoundException("object is not existed");
        }

        InputStream inputStream = object.downloadObjectAsInputStream();
        byte[] turnByte = getByte(inputStream);
        return new FSDataInputStream(new CephObjectInputStream(turnByte));
    }

    @Override
    public FSDataOutputStream create(Path path, FsPermission fsPermission,
        boolean overwrite, int bufferSize, short replication, long blockSize,
        Progressable progressable) throws IOException {
        CephObjectOutputStream out = this.createCephObjectOutputStream(path);
        return new FSDataOutputStream(out, this.statistics);
    }

    /**
     * append模式不支持
     * @param path 路径
     * @param i 从i开始追加
     * @return 不支持
     * @throws IOException exception
     */
    @Override
    public FSDataOutputStream append(Path path, int i, Progressable progressable) throws IOException {
        throw new IOException("append is not support for CephSystem");
    }

    /***
     * 重命名不支持
     * @param oldPath 旧路径
     * @param newPath 新路径
     * @return false
     * @throws IOException 不支持
     */
    @Override
    public boolean rename(Path oldPath, Path newPath) throws IOException {
        throw new IOException("rename is not support for CephSystem");
    }

    /**
     * 删除ceph对象平台对象
     * @param path ceph对象平台路径
     * @param deleteAll 是否删除全部
     * @return boolean
     * @throws IOException exception
     */
    @Override
    public boolean delete(Path path, boolean deleteAll) throws IOException {
        CephObjectAttributes cephFiLe = CephObjectUtil.getCephFiLeByPath(path);
        try {
            //获得桶
            if (null == cephFiLe.getBucketName()){
                return false;
            }
            Container container = account.getContainer(cephFiLe.getBucketName());
            if (null == cephFiLe.getObjectName()){
                if(container.exists()){
                    container.delete();
                }
            }else {
                String objectName = cephFiLe.getObjectName();
                StoredObject object = container.getObject(objectName);
                object.delete();
            }
            return true;
        } catch (Exception e) {
            LOG.error("删除:"+ path + " 失败");
        }
        return false;
    }

    @Override
    public FileStatus[] listStatus(Path path) throws IOException {
        List<CephObjectStatus> cephObjectStatuses = new ArrayList<>();
        CephObjectAttributes cephAttributes = CephObjectUtil.getCephFiLeByPath(path);
        String bucketName = cephAttributes.getBucketName();
        String objectName = cephAttributes.getObjectName();
        if (null != bucketName){
            Container container = account.getContainer(bucketName);
            if (!container.exists()){
                throw new IOException(bucketName+" 桶不存在");
            }
            //path只包含桶名
            if (null == objectName){
                Collection<DirectoryOrObject> all = container.listDirectory();
                for (DirectoryOrObject data : all) {
                    //是对象的情况
                    if (data.isObject()){
                        StoredObject asObject = data.getAsObject();
                        CephObjectStatus status = new CephObjectStatus(asObject.getContentLength(), false, 1, 0,
                                asObject.getLastModifiedAsDate().getTime(), new Path(path+"/"+asObject.getName()));
                        cephObjectStatuses.add(status);
                    }
                    //是dir的情况
                    if (data.isDirectory()){
                        Directory asDirectory = data.getAsDirectory();

                        CephObjectStatus status = new CephObjectStatus(0, true, 1, 0, 0,
                                new Path(path+"/"+asDirectory.getName()));
                        cephObjectStatuses.add(status);
                    }
                }
            }else {
                Collection<StoredObject> list = container.list(objectName, objectName, 10000);
                for (StoredObject object : list) {
                    CephObjectStatus status = new CephObjectStatus(object.getContentLength(), false, 1, 0, object.getLastModifiedAsDate().getTime(), new Path(path + "/" + object.getBareName()));
                    cephObjectStatuses.add(status);
                }
            }
        }
        return cephObjectStatuses.toArray(new CephObjectStatus[0]);
    }

    @Override
    public void setWorkingDirectory(Path path) {

    }

    @Override
    public Path getWorkingDirectory() {
        return new Path(uri.toString());
    }

    /**
     * 创建一个桶
     * @param path path
     * @param fsPermission 权限
     * @return boolean
     * @throws IOException not found
     */
    @Override
    public boolean mkdirs(Path path, FsPermission fsPermission) throws IOException {
        if (!path.toString().endsWith("/")){
            String newPath = path + "/";
            path = new Path(newPath);
        }
        //将path切割成桶名和对象名
        CephObjectAttributes cephFiLe = CephObjectUtil.getCephFiLeByPath(path);
        String bucketName = cephFiLe.getBucketName();
        try {
            if (null != bucketName) {
                Container container = account.getContainer(bucketName);
                //桶不存在则建立,桶存在直接返回true
                if (!container.exists()){
                    container.create();
                }
            }
            return true;
        } catch (Exception e) {
            LOG.error("创建:"+ path + " 失败");
        }
        return false;
    }

    @Override
    public FileStatus getFileStatus(Path path) throws IOException {
        //桶名
        CephObjectAttributes attributes = CephObjectUtil.getCephFiLeByPath(path);
        String bucketName = attributes.getBucketName();
        Container container = account.getContainer(bucketName);
        if (!container.exists()){
            throw new IOException("桶"+bucketName+ "不存在");
        }
        //对象不存在就当文件夹返回
        String objectName = attributes.getObjectName();
        if (null == objectName){
            return new CephObjectStatus(0,true,1,0,0,path);
        }
        StoredObject object = container.getObject(objectName);
        if (!object.exists()){
            return new CephObjectStatus(0,true,1,0,0,path);
        }
        long length = object.getContentLength();
        long time = object.getLastModifiedAsDate().getTime();
        return new CephObjectStatus(length,false,1,0,time,path);

    }

    /**
     * 将一个流转成字节数组
     * @param in 输入流
     * @return byte[]
     * @throws IOException 异常
     */
    private byte[] getByte(InputStream in) throws IOException {
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        byte[] buf = new byte[1024];
        int read;
        while ((read = in.read(buf)) > 0)
            outputStream.write(buf, 0, read);
        return outputStream.toByteArray();
    }

    protected CephObjectOutputStream createCephObjectOutputStream(Path path) throws IOException {
        //暂定1024
        long partSizeKB = 1024;
        return new CephObjectOutputStream(this.getConf(),path.toUri().toString(), partSizeKB, account);
    }

}
